package com.gs.netty.client.remoting;

import com.alibaba.fastjson2.JSON;
import com.gs.netty.client.*;
import com.gs.netty.client.enums.CommunicationMode;
import com.gs.netty.client.exception.*;
import com.gs.netty.client.util.ChannelUtils;
import com.gs.netty.core.ResponseCode;
import com.gs.netty.core.common.NettyConstants;

import com.gs.netty.support.LoadBalance;

import com.gs.netty.support.dispatch.WebSocketDispatch;
import com.gs.netty.support.codec.WebSocketAcceptMessageProtocolDecoder;
import com.gs.netty.support.codec.WebSocketSendMessageProtocolEncoder;

import com.gs.netty.websocket.remoting.WebSocketInvocation;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;
import javax.net.ssl.SSLEngine;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

@Slf4j
public class WebSocketClient extends AbstractRemotingClient {
    private volatile int reselectCount = 10;
    private float delayRate = 1.25f;
    private long delayMax = TimeUnit.MINUTES.toMillis(30);
    private final Bootstrap bootstrap  = new Bootstrap();
    private final EventLoopGroup workerGroup;


    private WebSocketClientDispatchHandler clientDispatchHandler;


    /**
     * 定时任务
     */
    private final PrevHashedWheelTimer requestResponseTableTimer = new PrevHashedWheelTimer(r -> new Thread(r, "requestResponseTableTimerService"));
    private final PrevHashedWheelTimer reconnectTimer = new PrevHashedWheelTimer(r -> new Thread(r, "reconnectTimerService"));
    /**
     * 负载均衡器
     */
//    private LoadBalance<ChannelWrapper> channelLoadBalance = new RoundRobinLoadBalance();


    // Channel 表锁
    private final Lock lockChannelTables = new ReentrantLock();
    /**
     *
     * 限制异步发送的并发数
     */
    protected final Semaphore semaphoreAsync;
    /**
     * 绑定的
     */
    private volatile ChannelWrapper stickyChannelWrapper;

    /**
     * rpc 调用使用
     */
    private final ConcurrentMap<Channel , ChannelWrapper> channelTables = new ConcurrentHashMap<>();
    /**
     * 创建重连使用
     * key：地址
     * value： Channel 包装对象
     */
    private final ConcurrentMap<String /* addr */, ChannelWrapper> addressChannelTables = new ConcurrentHashMap<>();

    protected boolean availableCheck =true;

    public WebSocketClient(NettyClientConfig nettyClientConfig, WebSocketDispatch webSocketDispatch) throws Exception {


        super(nettyClientConfig);
        workerGroup  = buildWorkerEventLoopGroupSelector();
        // 创建
        clientDispatchHandler = new WebSocketClientDispatchHandler(nettyClientConfig);
        clientDispatchHandler.setDispatch(webSocketDispatch);
        semaphoreAsync = new Semaphore(nettyClientConfig.getPermitsAsync(),true);
    }

    @Override
    protected Semaphore getSemaphoreAsync() {
        return semaphoreAsync;
    }

    @Override
    public void start() {
        NettyClientConfig nettyClientConfig = getBaseNettyConfig(NettyClientConfig.class);
        bootstrap.group(workerGroup) //设置线程组
                    //指定客服端通道实现类型NioSocketChannel
                    .channel(NioSocketChannel.class)
                    // 设置客户端连接超时时间
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis());


       // 项目初始化  Channel
       initChannelWrapper(nettyClientConfig);
        // 启动每隔1秒去扫描检查响应表
        io.netty.util.TimerTask timerTaskScanResponseTable = new io.netty.util.TimerTask() {
            @Override
            public void run(Timeout timeout) {
                try {
                    // 扫描检查响应表
                    WebSocketClient.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                } finally {
                    requestResponseTableTimer.newTimeout(this, 1000, TimeUnit.MILLISECONDS);
                }
            }
        };
        // 3秒后启动
        this.requestResponseTableTimer.newTimeout(timerTaskScanResponseTable, 1000 * 3, TimeUnit.MILLISECONDS);
    }


    @Override
    public void init() {

    }

    @Override
    public void shutdown() {
        if (closed.compareAndSet(false, true)) {
            for (ChannelWrapper channelWrapper : channelTables.values()) {

            }

            workerGroup.shutdownGracefully();
        }
    }


    protected ConcurrentHashMap<Integer,ResponseFuture> getResponseTables(){
        return clientDispatchHandler.getResponseTables();
    }

    /**
     * 获取发送超时时间
     * @return
     */
    @Override
    protected long getSendTimeout() {
        return getBaseNettyConfig(NettyClientConfig.class).getSendTimeout();
    }

    /**
     * 处理请求错误
     * @param requestId
     */
    @Override
    protected void requestFail(Integer requestId) {
        // 移除响应表
        ResponseFuture responseFuture = getResponseTables().remove(requestId);
        // 不等于 null
        if (responseFuture != null) {
            // 设置发送失败标准
            responseFuture.setSendRequestOK(false);
            // 设置空结果
            responseFuture.putResponse(null);
            try {
                // 执行回答函数
                clientDispatchHandler.executeInvokeCallback(responseFuture);
            } catch (Throwable e) {
                log.warn("execute callback in requestFail, and callback throw", e);
            } finally {
                responseFuture.release();
            }
        }
    }


    /**
     * 消息发送
     * @param message
     * @param mode
     * @param sendCallback
     * @param timeout
     * @return
     */
    @Override
    public <T> T doSend(Message message, CommunicationMode mode,
                        SendCallback sendCallback, long timeout,
                        Class<T> requiredType)
            throws WebSocketTooMuchRequestTimeoutException,
            WebSocketNotFountChannelException, WebSocketClientException {
        long beginStartTime = System.currentTimeMillis();

        long beginTimestampPrev = beginStartTime;

        WebSocketInvocation  rpcInvocation = new WebSocketInvocation(message.getCommand(),message.getBody());
        // 失败发送次数
        int timesTotal = mode == CommunicationMode.SYNC ? 1 + getBaseNettyConfig(NettyClientConfig.class).getRetryTimesWhenSendFailed() : 1;
        // 当前发送次数
        int times = 0;
        // 发送超时标准
        boolean callTimeout = false;
        Exception exception = null;
        ChannelWrapper channelWrapperPrev=null;

        /**
         * 需要轮询的 Channel
         */
        List<ChannelWrapper> invokerChannels = new ArrayList<>(channelTables.values());

        /**
         * 选举过的Channel
         */
        List<ChannelWrapper> selectedChannels = new ArrayList<>(invokerChannels.size());

        /**
         * channelWrapperList  和 selectChannelWrapperList 作用区别
         *
         *
         */
        SendResult sendResult = null;

        do {
            try {

                // 负载均衡 获取 channel
                channelWrapperPrev = select(selectedChannels,invokerChannels,rpcInvocation);

                if (channelWrapperPrev != null) {
                    selectedChannels.add(channelWrapperPrev);
                    beginTimestampPrev = System.currentTimeMillis();

                    long costTime = beginTimestampPrev - beginStartTime;
                    // 判断是否超时
                    if (timeout < costTime) {
                        // 设置调用超时标志
                        callTimeout = true;
                        break;
                    }
                    // 发送
                     sendResult = invoke(channelWrapperPrev,

                            mode, message,
                            timeout - costTime, sendCallback, requiredType);
                    // 判断调用类型是同步还是异步
                    switch (mode) {
                        case ASYNC:
                            return null;
                        case ONEWAY:
                            return null;
                        case SYNC:
                            return (T) sendResult;
                        default:
                            break;
                    }
                }else {
                    break;
                }
            } catch (WebSocketRemotingConnectException e) {
                exception = e;
            } catch (WebSocketNotFountChannelException e) {
                exception = e;
            } catch (WebSocketServerException e) {
                throw new RuntimeException(e);
            } catch (WebSocketRemotingSendRequestException e) {
                     exception = e;;
            } catch (WebSocketRemotingTimeoutException e) {
                     exception = e;;
            }catch (JsonConversionException e) {
              exception = e;;
            }  catch (WebSocketClientHandshakerException e) {
                exception = e;
            }
        }while (times++ < timesTotal);

        // 判断是否调用超时
        if (callTimeout) {
            throw new WebSocketTooMuchRequestTimeoutException("send timeout ");
        }
        if (exception != null) {
           throw   new WebSocketClientException(exception);
        }
        throw  new WebSocketNotFountChannelException("No route info of this Channel");
    }



    private void initChannelWrapper(NettyClientConfig nettyClientConfig) {
        log.info("init create websocket channel ...");
        for (String address :  nettyClientConfig.getWebSocketPaths()) {
            try {
                createChannel(address);
            } catch (Exception e) {
                log.error("init create websocket channel error address = {} errorMsg: {}",address,e.getMessage());
            }
        }
    }


    private void scanResponseTable() {

        // 移除的响应
        final List<ResponseFuture> rfList = new LinkedList<>();

        Iterator<Map.Entry<Integer, ResponseFuture>> it = clientDispatchHandler.getResponseTables().entrySet().iterator();

        // 遍历
        while (it.hasNext()) {
            Map.Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();
            /**
             * 调用开始时间 + 超时时间  小于 当前时间 证明是调用超时了
             */
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                rep.release();
                // 移除
                it.remove();
                rfList.add(rep);
                log.warn("remove timeout request, " + rep);
            }
        }

        for (ResponseFuture rf : rfList) {
            try {
                // 执行响应回调函数
                clientDispatchHandler.executeInvokeCallback(rf);
            } catch (Throwable e) {
                log.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }

    private Channel createChannel(String address) throws   WebSocketRemotingConnectException, WebSocketNotFountChannelException {

        ChannelWrapper channelWrapper = addressChannelTables.get(address);

        // 不为null 并且是有效的
        if (channelWrapper != null&& channelWrapper.isAvailable()) {
            return channelWrapper.getChannel();
        }
        try {
            if (this.lockChannelTables.tryLock(getBaseNettyConfig(NettyClientConfig.class).getLockTimeoutMillis(), TimeUnit.MILLISECONDS)) {
                try {
                    boolean createNewConnection;
                    channelWrapper = addressChannelTables.get(address);
                    if (channelWrapper != null) {
                        if (channelWrapper.isAvailable()) {
                            // 证明  channel 是没有问题的
                            return channelWrapper.getChannel();

                        } else if (!channelWrapper.getChannelFuture().isDone()) {// 证明还没完成等连接完成
                            createNewConnection = false;
                        } else {
                            // 证明确实没有连接
                            channelTables.remove(channelWrapper.getChannel());
                            addressChannelTables.remove(address);
                            createNewConnection = true;
                        }
                    }else {
                        createNewConnection = true;
                    }

                    if (createNewConnection) {
                        // 创建连接
                        URI uri = URI.create(address);
                        WebSocketClientChannelInitializer channelInitializer = new WebSocketClientChannelInitializer(uri);
                        ChannelFuture  channelFuture =  createConnect(channelInitializer);
                        channelWrapper = new ChannelWrapper(channelInitializer, channelFuture);
                        addressChannelTables.put(address,channelWrapper);
                        channelTables.put(channelFuture.channel(),channelWrapper);
                    }
                } catch (WebSocketRemotingConnectException e){
                    throw e;
                } catch (Exception e) {
                    throw e;
                } finally {
                    this.lockChannelTables.unlock();
                }
            }else {
                log.warn("createChannel: try to lock channel table, but timeout, {}ms", getBaseNettyConfig(NettyClientConfig.class).getLockTimeoutMillis());
            }
        } catch (InterruptedException e) {
            log.error("createChannel: create channel exception", e);
        }

        if (channelWrapper != null) {
            return waitChannelFuture(address, channelWrapper);
        }
        channelWrapper = addressChannelTables.get(address);

        if (channelWrapper != null && channelWrapper.isAvailable()) {
            // 证明  channel 是没有问题的
            return channelWrapper.getChannel();
        }

        throw  new WebSocketNotFountChannelException("Channel Not Fount");
    }

    private Channel waitChannelFuture(String addr, ChannelWrapper cw){
        ChannelFuture channelFuture = cw.getChannelFuture();
        // 连接等待
        if (channelFuture.awaitUninterruptibly(this.getBaseNettyConfig(NettyClientConfig.class).getConnectTimeoutMillis())) {
            // 判断是否可用
            if (cw.isAvailable()) {
                log.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                return cw.getChannel();
            } else {
                log.warn("createChannel: connect remote host[{}] failed, {}", addr, channelFuture.toString());
            }
        } else {
            log.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.getBaseNettyConfig(NettyClientConfig.class).getConnectTimeoutMillis(), channelFuture.toString());
        }
        return null;
    }

    /**
     * 连接
     * @param channelInitializer
     * @return
     */
    private ChannelFuture createConnect(WebSocketClientChannelInitializer channelInitializer) throws WebSocketRemotingConnectException, InterruptedException {
        // 创建连接

        return bootstrap.handler(channelInitializer).connect(channelInitializer.getUri().getHost(), channelInitializer.getUri().getPort()).sync();

    }



    public void closeChannel(Channel channel) {
        if (null == channel) {
            return;
        }

        try {
            if (this.lockChannelTables.tryLock(getBaseNettyConfig(NettyClientConfig.class).getLockTimeoutMillis(), TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    ChannelWrapper prevCW = null;
                    String addrRemote = null;

                    for (Map.Entry<String, ChannelWrapper> entry : addressChannelTables.entrySet()) {
                        String key = entry.getKey();
                        ChannelWrapper prev = entry.getValue();
                        if (prev.getChannel() != null) {

                            if (prev.getChannel() == channel) {
                                prevCW = prev;
                                addrRemote = key;
                                break;
                            }
                        }
                    }
                    if (null == prevCW) {
                        log.info("eventCloseChannel: the channel[{}] has been removed from the channel table before", addrRemote);
                        removeItemFromTable = false;
                    }
                    if (removeItemFromTable) {
                        // 移除
                        this.channelTables.remove(addrRemote);
                        log.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                        // 关闭连接
                        ChannelUtils.closeChannel(channel);
                    }
                } catch (Exception e) {
                    log.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                log.warn("closeChannel: try to lock channel table, but timeout, {}ms", getBaseNettyConfig(NettyClientConfig.class).getLockTimeoutMillis());
            }
        } catch (InterruptedException e) {
            log.error("closeChannel exception", e);
        }
    }

    protected <T> SendResult<T> invoke(final ChannelWrapper channelWrapper,
                                                       CommunicationMode mode,
                                                       Message message,
                                                       final long timeout,
                                SendCallback sendCallback,
                                Class<T> requiredType) throws WebSocketTooMuchRequestTimeoutException, WebSocketRemotingConnectException,
            WebSocketServerException, WebSocketRemotingSendRequestException,
            WebSocketRemotingTimeoutException, JsonConversionException, WebSocketClientHandshakerException {

        long beginStartTime = System.currentTimeMillis();
        WebSocketInvocation rpcInvocation = new WebSocketInvocation(message.getCommand(),message.getBody());
        switch (mode) {
            // 单向消息
            case ONEWAY:
                return null;
            // 异步
            case ASYNC:
                return null;
            // 同步
            case SYNC:
                long costTimeSync = System.currentTimeMillis() - beginStartTime;
                if (timeout < costTimeSync) {
                    throw new WebSocketTooMuchRequestTimeoutException("send timeout");
                }
                // 消息发送
                return this.invokeSync(channelWrapper, rpcInvocation, timeout - costTimeSync,requiredType);
            default:
                assert false;
                break;
        }
        return null;
    }

    private <T> SendResult<T> invokeSync(ChannelWrapper channelWrapper,
                                         WebSocketInvocation invocation,
                                         long timeoutMillis, Class<T> requiredType) throws WebSocketServerException,
            WebSocketRemotingConnectException, WebSocketRemotingSendRequestException,
            WebSocketRemotingTimeoutException, JsonConversionException, WebSocketClientHandshakerException {

        long beginStartTime = System.currentTimeMillis();

        Channel channel = channelWrapper.getChannel();

        String channelRemoteAddr =channelWrapper.getChannelAddress();

        if (channelWrapper.isAvailable()) {
            long left = timeoutMillis;
            long costTime = System.currentTimeMillis() - beginStartTime;

            left -= costTime;

            if (left <= 0) {
                throw new WebSocketRemotingTimeoutException("invokeSync call the addr[" + channelRemoteAddr + "] timeout");
            }
            if (channelWrapper.isHandshakeComplete()) {
                WebSocketInvocation response = invokeSync(channel, invocation, left);
                assert response != null;
                return processSendResponse(response,requiredType);
            }
            // 抛出未完成握手
            throw new WebSocketClientHandshakerException("un complete handshake");
        }else {
            this.closeChannel(channel);
            throw new WebSocketRemotingConnectException(channelRemoteAddr);
        }

    }


    protected WebSocketInvocation invokeSync(final Channel channel, final WebSocketInvocation rpcInvocation,
                                          final long timeoutMillis)
            throws WebSocketRemotingTimeoutException, WebSocketRemotingSendRequestException {

        // 远程调用
        WebSocketInvocation resp = invokeSync0(channel, rpcInvocation, timeoutMillis);
        // 更新最后响应消息
        updateChannelLastResponseTime(channel);
        return resp;
    }

    private void updateChannelLastResponseTime(Channel channel) {
        ChannelWrapper channelWrapper = this.channelTables.get(channel);

        if (channelWrapper != null && channelWrapper.isAvailable()) {
            channelWrapper.updateLastResponseTime();
        }

    }

    private <T> SendResult<T> processSendResponse(WebSocketInvocation response, Class<T> requiredType) throws WebSocketServerException, JsonConversionException {
        int status = response.getStatus();

        T data = null;

        int code =ResponseCode.OK;
        String message = null;
        switch (status) {
            case ResponseCode.OK:
                try {
                    if (requiredType!=null){
                        data = JSON.parseObject(response.getPayload(), requiredType);
                    }else {
                        data = (T) response.getPayload();
                    }
                } catch (Exception e) {
                    // 结果 json 转换 错误
                    throw new JsonConversionException("Result json conversion to "+requiredType.toString()+" error ",e);
                }
                break;
            case  ResponseCode.NO_FOUND_HANDLER:
                code = ResponseCode.NO_FOUND_HANDLER;
                message =  response.getPayload();
                break;
            default:
                throw new WebSocketServerException(String.valueOf(code),new String(response.getPayload()));
        }

        SendResult<T> sendResult = new SendResult();
        sendResult.setCode(code);
        sendResult.setMessage(message);
        sendResult.setData(data);
        return sendResult;
    }





    /**
     * 选择 channel
     * @param selectedChannels
     * @param invokerChannels
     * @param rpcInvocation
     * @return
     * @throws WebSocketRemotingConnectException
     * @throws WebSocketNotFountChannelException
     */
    private ChannelWrapper select(List<ChannelWrapper> selectedChannels,

                                                List<ChannelWrapper> invokerChannels,
                                                WebSocketInvocation rpcInvocation)  throws WebSocketRemotingConnectException, WebSocketNotFountChannelException {



        // 获取负载均衡器
        LoadBalance<ChannelWrapper> loadBalance = initLoadBalance(invokerChannels, rpcInvocation);

        // 负载均衡
        return select(loadBalance,selectedChannels, invokerChannels, rpcInvocation);
    }


    protected LoadBalance<ChannelWrapper> initLoadBalance(List<ChannelWrapper> invokers, WebSocketInvocation invocation) {

        return new RoundRobinLoadBalance();
    }

    private ChannelWrapper select(LoadBalance<ChannelWrapper> loadBalance,
                                  List<ChannelWrapper> selectedChannels,
                                  List<ChannelWrapper> invokerChannels,
                                  WebSocketInvocation invocation) {

        if (CollectionUtils.isEmpty(invokerChannels)) {
            reconnect();
            return null;
        }

        /**
         * 如果集群容错类 绑定了一个 Channel 并且 没有包含在缓存列表里面的，那么就移除掉
         */
        if (stickyChannelWrapper != null && !invokerChannels.contains(stickyChannelWrapper)) {

            stickyChannelWrapper = null;
        }

        /**
         * 如果集群容错类 绑定了一个 Channel，并且没有使用过，那么就直接是这个 Channel，不用去 负载均衡
         */
        if (getBaseNettyConfig(NettyClientConfig.class).isStickyChannel() && stickyChannelWrapper != null

                && (selectedChannels == null || !selectedChannels.contains(stickyChannelWrapper))) {
            if (availableCheck && stickyChannelWrapper.isAvailable()) {
                // 使用绑定的 Channel
                return stickyChannelWrapper;
            }
        }



        // 进行负载均衡算法选择
        ChannelWrapper  invokerChannel = doSelect(loadBalance,selectedChannels, invokerChannels, invocation);

        if (invokerChannel == null||!invokerChannel.isAvailable()) {
            reconnect();
        }
        if (getBaseNettyConfig(NettyClientConfig.class).isStickyChannel()) {
            stickyChannelWrapper = invokerChannel;
        }

        return invokerChannel;
    }

    private void  reconnect(){
        // 启动每隔1秒去扫描检查响应表
        io.netty.util.TimerTask timerTaskReconnect = new io.netty.util.TimerTask() {
            @Override
            public void run(Timeout timeout) {


                List<Throwable> exceptionList = new ArrayList<>();
                ArrayList<String> addArr = new ArrayList<>();
                try {
                    for (String addr : getBaseNettyConfig(NettyClientConfig.class).getWebSocketPaths()) {
                        try {
                            createChannel(addr);
                        } catch (Exception e) {
                            exceptionList.add(e);
                            addArr.add(addr);
                        }
                    }
                } finally {

                    if (!CollectionUtils.isEmpty(exceptionList)) {
                        PrevHashedWheelTimer.HashedWheelTimeout prevHashedWheelTimeout  = (PrevHashedWheelTimer.HashedWheelTimeout) timeout;
                        try {
                            // 上一个任务的毫秒
                            long preDelay = prevHashedWheelTimeout.delay();
                            // 下一个任务的毫秒
                            long  nextDelay = (long) (preDelay * delayRate);

                            if (nextDelay > delayMax) {// 最大延迟 30分钟
                                nextDelay = delayMax;
                            }

                            log.error("reconnect websocket channel error , preDelay {} , nextDelay {} , address = {}  ,errorMsg: {}",preDelay,nextDelay,addArr,exceptionList.stream().map(Throwable::getMessage).collect(Collectors.toSet()));

                            reconnectTimer.newTimeout(this,nextDelay, TimeUnit.MILLISECONDS);
                        } catch (Exception e) {

                        }


                    }

                }
            }
        };
        // 2秒后启动
        this.reconnectTimer.newTimeout(timerTaskReconnect, 2000, TimeUnit.MILLISECONDS);
    }
    private ChannelWrapper doSelect(LoadBalance<ChannelWrapper> loadBalance, List<ChannelWrapper> selectedChannels,
                                    List<ChannelWrapper> invokerChannels,
                                    WebSocketInvocation invocation) {

        if (CollectionUtils.isEmpty(invokerChannels)) {
            return null;
        }

        if (invokerChannels.size() == 1) {

            ChannelWrapper channelWrapper  = invokerChannels.get(0);
            // 校验是否有效
            checkShouldInvalidateInvoker(channelWrapper);
            return channelWrapper;
        }


        // 负载均衡
        ChannelWrapper invoker = loadBalance.loadBalance(invokerChannels);

        // 判断负载均衡 选择出来的 invoker 是否存在已经使用过的 selectedChannels 集合中
        boolean isSelected = selectedChannels != null && selectedChannels.contains(invoker);
        // 检查是否可用
        boolean isUnavailable = availableCheck && !invoker.isAvailable()&& selectedChannels.add(invoker);
        // 当前 负载均衡 出来的 invoker 是已经调用过了 或者 是这个 invoker 不可用了 ，那么这种情况就要重新选举 rInvoker
        if (isSelected ||isUnavailable) {
            try {
                // 重新选举
                ChannelWrapper rInvoker = reselect(loadBalance, invocation, invokerChannels, selectedChannels, availableCheck);

                // 判断重新选举的是否不为 null
                if (rInvoker != null) {
                    // 不为null，那么就使用 选举 invoker
                    invoker = rInvoker;
                } else {
                    // 选举的 invoker 为 null

                    //Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
                    // 获取 上一个使用过的 invoker 索引
                    int index = invokerChannels.indexOf(invoker);

                    try {

                        //Avoid collision
                        // 上一个使用过的 invoker 索引 +1 获取 invoker
                        invoker = invokerChannels.get((index + 1) % invokerChannels.size());

                    } catch (Exception e) {
//                        log.warn(CLUSTER_FAILED_RESELECT_INVOKERS,"select invokers exception","",e.getMessage() + " may because invokers list dynamic change, ignore.",e);
                    }
                }
            } catch (Throwable t) {
//                logger.error(CLUSTER_FAILED_RESELECT_INVOKERS,"failed to reselect invokers","","cluster reselect fail reason is :" + t.getMessage() + " if can not solve, you can set cluster.availablecheck=false in url",t);
            }
        }
        return invoker;
    }

    private ChannelWrapper reselect(LoadBalance<ChannelWrapper> loadBalance,
                                    WebSocketInvocation rpcInvocation,
                                    List<ChannelWrapper> invokerChannels,
                                    List<ChannelWrapper> selectedChannels, boolean availableCheck) {

        // 重新选举的 Invoker 列表
        List<ChannelWrapper> reselectInvokers = new ArrayList<>(Math.min(invokerChannels.size(), reselectCount));


        // 重新选择数 大于 提供者 invoker 数量 没有大于默认
        if (reselectCount >= invokerChannels.size()) {

            for (ChannelWrapper invoker : invokerChannels) {
                // check if available
                // 如果不可用 就跳过
                if (availableCheck && !invoker.isAvailable()) {
                    // add to invalidate invoker
//                    invalidateInvoker(invoker);
                    continue;
                }

                // 如果该 invoker 不在已使用过的 invoker 列表 那么该 invoker 就会添加到 重新选举 Invoker 列表中 进行选举
                if (selectedChannels == null || !selectedChannels.contains(invoker)) {
                    // 添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }
        } else {
            // 如果 没有大于

            for (int i = 0; i < reselectCount; i++) {

                // select one randomly
                // 随机获取 一个
                ChannelWrapper invoker = invokerChannels.get(ThreadLocalRandom.current().nextInt(invokerChannels.size()));


                // check if available
                // 如果不可用 就跳过
                if (availableCheck && !invoker.isAvailable()) {

                    // add to invalidate invoker
                    invalidateInvoker(invoker);
                    continue;
                }

                // de-duplication
                // 判断这个 invoker 是否存在调用过了，或者存在随机获取重复了
                if (selectedChannels == null || !selectedChannels.contains(invoker) || !reselectInvokers.contains(invoker)) {
                    // 添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }

        }

        // 2. Use loadBalance to select one (all the reselectInvokers are available)
        // 重新选举 Invoker 列表 不为null，那么进行负载均衡
        if (!reselectInvokers.isEmpty()) {
            // 重新负载均衡
            return loadBalance.loadBalance(reselectInvokers);
        }

        // 3. reselectInvokers is empty. Unable to find at least one available invoker.
        //    Re-check all the selected invokers. If some in the selected list are available, add to reselectInvokers.
        /**
         *  重新选举 Invoker 列表 为null，那么从已使用过的 Invoker 列表 找到可用的 Invoker 添加进 重新选举 Invoker 列表
         */
        if (selectedChannels != null) {

            for (ChannelWrapper invoker : selectedChannels) {

                if ((invoker.isAvailable()) // available first
                        && !reselectInvokers.contains(invoker)) {
                    // 可用，添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }
        }

        // 4. If reselectInvokers is not empty after re-check.
        //    Pick an available invoker using loadBalance policy

        // 重新选举 Invoker 列表 不为null，那么进行负载均衡
        if (!reselectInvokers.isEmpty()) {

            return loadBalance.loadBalance(reselectInvokers);
        }
        // 5. No invoker match, return null.
        // 代码执行到这，证明确实是 没有可用的 invoker 了
        return null;
    }

    private void invalidateInvoker(ChannelWrapper invoker) {

    }

    private void checkShouldInvalidateInvoker(ChannelWrapper channelWrapper) {

    }


  public class WebSocketClientChannelInitializer extends ChannelInitializer<SocketChannel>{

        private final URI uri;

        public URI getUri() {
          return uri;
        }

        public WebSocketClientChannelInitializer(URI uri) {
           this.uri = uri;
       }
       @Override
       protected void initChannel(SocketChannel ch) throws Exception {
           ChannelPipeline pipeline = ch.pipeline();
           if (isOpenSsl()) {
               SSLEngine sslEngine = getSslContext().createSSLEngine();
               sslEngine.setUseClientMode(getBaseNettyConfig(NettyClientConfig.class).isUseClientMode());
               sslEngine.setNeedClientAuth(getBaseNettyConfig(NettyClientConfig.class).isNeedClientAuth());
               SslHandler sslHandler = new SslHandler(sslEngine);
               pipeline.addLast("sslHandler",sslHandler);
               log.info("Add Prepend SSL handler");
           }else {
               log.warn("Connections are insecure as SSLContext is null!");
           }
           /**
            * 使用 websocket 为什么要添加 HttpServerCodec，HttpObjectAggregator这两个处理器？
            * 因为 websocket 建立连接时候会发一次 http 握手请求。握手成功了之后会从客户端的 SocketChannel移除这个两个处理器的
            * HttpServerCodec 是用来解析 http 协议
            *
            * HttpObjectAggregator
            */
           pipeline.addLast("httpClientCodec",new HttpClientCodec());
           pipeline.addLast("httpObjectAggregator",new HttpObjectAggregator(65536));
           ch.pipeline().addLast(new ChunkedWriteHandler());  // 处理大数据流
           pipeline.addLast("idleState",new IdleStateHandler(0,5,0));

           // 添加服务权重处理器
           pipeline.addLast(NettyConstants.AUTH_HANDLER,new ChannelInboundHandlerAdapter(){

               @Override
               public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                   if (msg instanceof  FullHttpResponse) {

                       FullHttpResponse httpMsg = (FullHttpResponse) msg;
                       // 设置请求头
                       String weight = httpMsg.headers().get(NettyConstants.SERVICE_WEIGHT);

                       if (StringUtils.isNotEmpty(weight)) {

                           ChannelWrapper channelWrapper = channelTables.get(ctx.channel());
                           // 设置权重
                           channelWrapper.setWeight(Integer.parseInt(weight));

                           // 下一个处理器处理
                           ctx.fireChannelRead(msg);
                           // 移除当前处理器
                           ctx.pipeline().remove(this);

                       }else {
                           ctx.fireChannelRead(msg);
                       }


                   }else {
                       ctx.fireChannelRead(msg);
                   }
               }
           });
           /**
            * 创建 WebSocket 握手处理器
            */
           DefaultHttpHeaders entries = new DefaultHttpHeaders();
           entries.set("Authorization","1111");
           WebSocketClientHandshaker webSocketClientHandshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13,
                   getBaseNettyConfig(NettyClientConfig.class).getSubProtocols(), true,entries );
           WebSocketClientProtocolHandler webSocketClientProtocolHandler = new WebSocketClientProtocolHandler(webSocketClientHandshaker, getBaseNettyConfig(NettyClientConfig.class).getHandshakeTimeoutMillis());
           pipeline.addLast(WebSocketClientProtocolHandler.class.getName(),webSocketClientProtocolHandler);
           webSocketClientProtocolHandler.handshaker().isHandshakeComplete();

           pipeline.addLast(new WebSocketFrameAggregator(10 * 1024 * 1024));


           pipeline.addLast(new WebSocketSendMessageProtocolEncoder());
           pipeline.addLast(new WebSocketAcceptMessageProtocolDecoder());
           // 添加客户端消息分发器
           pipeline.addLast(clientDispatchHandler.getClass().getName(),clientDispatchHandler);
       }
   }
}
